# -*- coding: utf-8 -*-
from utils.operators.cluster_for_spark_sql_operator import SparkSqlOperator

from datetime import timedelta
from jms.dm.dm_taking_scan_miss_detail import dm__dm_taking_scan_miss_detail
from jms.dim.tab.dim_t_ass_sorting_dict_base import jms_dim__dim_t_ass_sorting_dict_base
from jms.dim.dim_sys_agency_district_city_base import jms_dim__dim_sys_agency_district_city_base
from jms.dim.ide.dim_jds_comatch_configuration_base import jms_dim__dim_jds_comatch_configuration_base
# 揽收漏扫汇总
dm__dm_taking_scan_miss_rate = SparkSqlOperator(
    task_id='dm__dm_taking_scan_miss_rate',
    pool_slots=4,
    task_concurrency=1,  # 如果任务不支持并发，则将 task_concurrency 设为 1
    name='dm__dm_taking_scan_miss_rate_{{ execution_date | date_add(1) | cst_ds }}',
    sql='jms/dm/dm_taking_scan_miss_rate/execute.sql',
    driver_memory='3G' , 
    executor_memory='2G' , 
    executor_cores=2 , 
    num_executors=2 , 
    email=['guoruiling@jtexpress.com','yl_bigdata@yl-scm.com'],
    conf={
        'spark.dynamicAllocation.enabled': 'true',  # 动态资源开启
        'spark.shuffle.service.enabled': 'true',  # 动态资源 Shuffle 服务开启
        'spark.dynamicAllocation.maxExecutors': 4,  # 动态资源最大扩容 Executor 数
        'spark.executor.memoryOverhead': '2048' ,
        'spark.core.connection.ack.wait.timeout': 5000,
        'spark.default.parallelism': 20,
        'spark.locality.wait': 60,
        'spark.driver.maxResultSize': 8,
        'spark.hadoop.hive.exec.dynamic.partition.mode': 'nonstrict', # 动态分区
        'spark.hadoop.hive.exec.dynamic.partition': 'true',
         },
    execution_timeout=timedelta(minutes=30),
    hiveconf={'hive.exec.dynamic.partition': 'true',
              'hive.exec.dynamic.partition.mode': 'nonstrict',
              'hive.exec.max.dynamic.partitions.pernode': 200,
              'hive.exec.max.dynamic.partitions': 200,
              },
    yarn_queue='pro',
)

dm__dm_taking_scan_miss_rate << [
    dm__dm_taking_scan_miss_detail,
    jms_dim__dim_t_ass_sorting_dict_base,
    jms_dim__dim_sys_agency_district_city_base,
    jms_dim__dim_jds_comatch_configuration_base
]
